package com.bitdf.txing.chat.scheduled;

import com.alibaba.fastjson.JSON;
import com.bitdf.txing.chat.constant.Constant;
import com.bitdf.txing.chat.entity.ChatMessageEntity;
import com.bitdf.txing.chat.service.ChatMessageService;
import com.bitdf.txing.common.utils.TimeConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;


/**
 * 定时任务
 * 1、@EnableScheduling 开启定时任务
 * 2、@Scheduled开启一个定时任务
 * <p>
 * 异步任务
 * 1、@EnableAsync:开启异步任务
 * 2、@Async：给希望异步执行的方法标注
 */

@Slf4j
@Component
@EnableAsync
@EnableScheduling
public class ChatScheduled {

    @Autowired
    StringRedisTemplate redisTemplate;
    @Autowired
    ChatMessageService chatMessageService;

    /**
     * 凌晨三点 定时更新数据库（把redis中的聊天记录同步到数据库）
     *
     * @throws Exception
     */
    @Async
    @Scheduled(cron = "0 0 3 * * *")
    @Transactional
    public void updateData() throws Exception {
        log.info("[定时任务开始执行] ==> redis 同步聊天记录到 mysql");
        redisTemplate.execute((RedisCallback) connection -> {
            log.info("开始持久化用户私聊记录");
            try (Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions()
                    .match(Constant.CHATMESSAGE_SINGLE + "*")
                    .count(10000).build())) {
                while (cursor.hasNext()) {
                    String s = new String(cursor.next(), "Utf-8");
                    String substring = s.substring(s.lastIndexOf(":"));
                    log.info("遍历：{}", s);
                    // 查出此对用户在redis的所有聊天
                    Long count = redisTemplate.opsForList().size(s);
                    List<String> range = redisTemplate.opsForList().range(s, 0, count);
                    if (range == null || range.size() == 0) {
                        continue;
                    }
                    // 查出该对用户在数据库中最后一条消息的时间
                    String time = redisTemplate.opsForValue().get(Constant.LSAT_SAVE_MESSAGE_TIME + substring);
                    Date date = time == null ? TimeConstant.zeroDate : new Date(Long.parseLong(time));
                    // 查出该对用户在redis中新消息
                    List<ChatMessageEntity> collect = range.stream().map((str) -> {
                        ChatMessageEntity chatMessageEntity = JSON.parseObject(str, ChatMessageEntity.class);
                        return chatMessageEntity;
                    }).filter((item) -> {
                        return date.compareTo(item.getCreateTime()) < 0;
                    }).collect(Collectors.toList());
                    if (collect != null || collect.isEmpty()) {
                        log.info("用户对：{} 没有新的聊天记录 无需持久化", substring);
                        continue;
                    }
                    // 保存到数据库
                    boolean b = chatMessageService.saveBatch(collect);
                    // 删除redis中部分元素 保留20条 之所以不全删了： 是因为要预防太多用户一起去数据库查聊天记录 导致数据库压力过大
                    Long size = redisTemplate.opsForList().size(s);
                    redisTemplate.opsForList().trim(s, 0, 20);
                    // 更新时间
                    redisTemplate.opsForValue().set(Constant.LSAT_SAVE_MESSAGE_TIME + substring, JSON.toJSONString(new Date()));
                    log.info("用户对：{} 私聊信息已成功持久化", s);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                throw new RuntimeException(e);
            }

            log.info("开始持久化用户群聊记录");
            try (Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions()
                    .match(Constant.CHATMESSAGE_GROUP + "*")
                    .count(10000).build())) {
                while (cursor.hasNext()) {
                    String s = new String(cursor.next(), "Utf-8");
                    String substring = s.substring(s.lastIndexOf(":"));
                    log.info("遍历：{}", s);
                    // 查出此对用户在redis的所有聊天
                    Long count = redisTemplate.opsForList().size(s);
                    List<String> range = redisTemplate.opsForList().range(s, 0, count);
                    if (range == null || range.size() == 0) {
                        continue;
                    }
                    // 查出该对用户在数据库中最后一条消息的时间
                    String time = redisTemplate.opsForValue().get(Constant.LSAT_SAVE_MESSAGE_TIME + substring);
                    Date date = time == null ? TimeConstant.zeroDate : new Date(Long.parseLong(time));
                    // 查出该对用户在redis中新消息
                    List<ChatMessageEntity> collect = range.stream().map((str) -> {
                        ChatMessageEntity chatMessageEntity = JSON.parseObject(str, ChatMessageEntity.class);
                        return chatMessageEntity;
                    }).filter((item) -> {
                        return date.compareTo(item.getCreateTime()) < 0;
                    }).collect(Collectors.toList());
                    if (collect != null || collect.isEmpty()) {
                        log.info("用户对：{} 没有新的聊天记录 无需持久化", substring);
                        continue;
                    }
                    // 保存到数据库
                    boolean b = chatMessageService.saveBatch(collect);
                    // 删除redis中部分元素 保留20条 之所以不全删了： 是因为要预防太多用户一起去数据库查聊天记录 导致数据库压力过大
                    Long size = redisTemplate.opsForList().size(s);
                    redisTemplate.opsForList().trim(s, 0, 20);
                    // 更新时间
                    redisTemplate.opsForValue().set(Constant.LSAT_SAVE_MESSAGE_TIME + substring, JSON.toJSONString(new Date()));
                    log.info("用户对：{} 私聊信息已成功持久化", s);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                throw new RuntimeException(e);
            }
            return null;
        });
    }



}
